package org.ykhl.mq.commons.receiver;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.errors.WakeupException;
import org.ykhl.mq.commons.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by zkf on 2015/11/25.
 */
public class MessageReceiverKafkaImpl extends MessageReceiver {
    private static final Logger LOG = LoggerFactory.getLogger(MessageReceiverKafkaImpl.class);


    private ExecutorService executorService;
    private StoppableInfiniteRunnable[] runners;

    public MessageReceiverKafkaImpl(String serverAddress, String authKey, String securityFilePath,
                                    Constants.Role role, String topicPrefix,
                                    String consumerPrefix, MessageListener listener) {
        super(serverAddress, authKey, securityFilePath, role, topicPrefix, consumerPrefix, listener);

       // this.runner = new Runner();
    }

    public MessageReceiverKafkaImpl(String serverAddress, String authKey, String securityFilePath,
                                    Constants.Role role, MessageListener listener) {
        super(serverAddress, authKey, securityFilePath, role, listener);

        //this.runner = new Runner();
    }

    public MessageReceiverKafkaImpl(String serverAddress, String authKey, String securityFilePath,
                                    Constants.Role role) {
        this(serverAddress, authKey, securityFilePath, role, null);
    }

    @Override
    protected void doStart() throws Exception {
        this.executorService = Executors.newCachedThreadPool(
                new NamedThreadFactory("MessageFetcher"));

        runners = new StoppableInfiniteRunnable[1];
        for(int i=0;i<1;i++){
              runners[i] = new Runner();
            this.executorService.submit(runners[i]);
        }
//
//        this.executorService.submit(runner);
    }

    @Override
    protected void doStop() throws Exception {
        for (int i =0 ;i<runners.length; i++){
            runners[i].stop();
        }
//        runner.stop();
        executorService.shutdown();
    }

    private class Runner extends StoppableInfiniteRunnable {
        private Consumer<String, String> consumer = null;
        @Override
        public void stop(){
            super.stop();
            consumer.wakeup();
        }


        @Override
        public void beforeRun() {
            consumer = new KafkaConsumer<>(Utils.genKafkaConsumerProps(
                    serverAddress, consumerGroupName, useSsl, securityFilePath));
            consumer.subscribe(Collections.singletonList(topic));
        }

        @Override
        public void doRun() {
            try {
                ConsumerRecords<String, String> records = consumer.poll(2000);

                for (ConsumerRecord<String, String> record : records) {
                    LOG.debug("get " + records.count() + " messages at one time");
                    try {
                        if (consume(new BaseMessageKafkaImpl(record)) == MessageConsumeStatus.RECONSUME_LATER) {
                            LOG.warn("Kafka doesn't support RECONSUME_LATER");    //TODO how to solve?
                        }

                    } catch (Exception e) {
                        LOG.error("exception thrown from user's listener", e);
                    }
                }
                consumer.commitAsync();
//                if (records != null && !records.isEmpty())
//                    consumer.commitSync();
            } catch (WakeupException e){
                // wake up by other threads, going to stop, do nothing, just recheck state will do
                LOG.error("WakeupException", e);
            } catch (CommitFailedException e) {
                LOG.error("failed to commit, commit next time then", e);
                consumer.commitSync();
            } catch (Exception e) {
                LOG.error("unexpected error! ", e);
            }
        }

        @Override
        public void afterRun() {
            consumer.close();
        }
    }
}
